package com.study.bloomFilter.redisBloom.bitmap;

import java.nio.charset.Charset;
import org.apache.log4j.Logger;
import com.google.common.hash.Funnels;
import com.google.common.hash.Hashing;
import com.google.common.primitives.Longs;
import com.study.bloomFilter.error.BloomFilterException;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

/**
 * 
 * @描述: 使用redis bitmap实现布隆过滤器
 * @版权: Copyright (c) 2020 
 * @公司: 
 * @作者: 严磊
 * @版本: 1.0 
 * @创建日期: 2020年5月31日 
 * @创建时间: 上午11:06:07
 */
public class RedisBloomFilter
{
    private static final Logger LOGGER        = Logger.getLogger(RedisBloomFilter.class);
    
    private static final String BF_KEY_PREFIX = "bf:";
    
    private int                 numHashFunctions;                                        //hash函数个数
    
    private long                bitmapLength;                                            //bit位数
    
    private Jedis               jedis;                                                   //redis连接池
    
    /**
     *	描述： 构造方法，构造布隆过滤器
     * 	@param numApproxElements    预估元素数量
     * 	@param fpp                  可接受的最大误差（假阳性率）
     * 	@param jedisResourcePool    Codis专用的Jedis连接池
     */
    public RedisBloomFilter(long numApproxElements, double fpp, Jedis jedis)
    {
        this.jedis = jedis;
        
        if ( fpp >= 1 || fpp <= 0 )
        {
            new RuntimeException("可接受的最大误差必须满足0<fpp<1");
        }
        //限制bitmapLength大为4294967296， redis bitMap映射被限制在 512MB之内，最多存储2^32（ 4294967296）数据
        bitmapLength = Math.min((long) ( -numApproxElements * Math.log(fpp) / (Math.log(2) * Math.log(2))), 2L << 31);
        
        numHashFunctions = Math.max(1, (int) Math.round((double) bitmapLength / numApproxElements * Math.log(2)));
    }
    
    /**
     * 
     * @描述：取得自动计算的最优哈希函数个数
     * @作者：严磊
     * @时间：2020年5月31日 上午11:26:36
     * @return
     */
    public int getNumHashFunctions()
    {
        return numHashFunctions;
    }
    
    /**
     * 
     * @描述：取得自动计算的最优Bitmap长度
     * @作者：严磊
     * @时间：2020年5月31日 上午11:26:30
     * @return
     */
    public long getBitmapLength()
    {
        return bitmapLength;
    }
    
    /**
     * 
     * @描述：计算一个元素值哈希后映射到Bitmap的哪些bit上
     *      使用的是 com.google.common.hash.BloomFilterStrategies MURMUR128_MITZ_64() 
     * @作者：严磊
     * @时间：2020年5月31日 上午11:26:05
     * @param element   元素值
     * @return          bit下标的数组
     */
    private long[] getBitIndices(String element)
    {
        long[] indices = new long[numHashFunctions];
        
        byte[] bytes = Hashing.murmur3_128().hashObject(element, Funnels.stringFunnel(Charset.forName("UTF-8")))
                .asBytes();
        
        long hash1 = Longs.fromBytes(bytes[7], bytes[6], bytes[5], bytes[4], bytes[3], bytes[2], bytes[1], bytes[0]);
        long hash2 = Longs.fromBytes(bytes[15], bytes[14], bytes[13], bytes[12], bytes[11], bytes[10], bytes[9],
                bytes[8]);
        
        long combinedHash = hash1;
        for (int i = 0; i < numHashFunctions; i++)
        {
            indices[i] = (combinedHash & Long.MAX_VALUE) % bitmapLength;
            combinedHash += hash2;
        }
        
        return indices;
        
    }
    
    /**
     * 
     * @描述：插入元素
     * @作者：严磊
     * @时间：2020年5月31日 上午11:25:27
     * @param key       原始Redis键，会自动加上'bf:'前缀
     * @param element   元素值，字符串类型
     * @param expireSec 过期时间（秒）
     */
    public void addElement(String key, String element, int expireSec)
    {
        if ( key == null || element == null )
        {
            throw new RuntimeException("键值均不能为空");
        }
        String actualKey = BF_KEY_PREFIX.concat(key);
        /**
         * 使用redis 管道Pipeline，将命令打包推送 参考：https://www.cnblogs.com/yepei/p/5662734.html
         */
        try (Pipeline pipeline = jedis.pipelined())
        {
            for (long index : getBitIndices(element))
            {
                pipeline.setbit(actualKey, index, true);
            }
            pipeline.syncAndReturnAll();
            jedis.expire(actualKey, expireSec);
        }
        catch (Exception ex)
        {
            LOGGER.error("资源关闭时发生Exception", ex);
            throw new BloomFilterException(ex);
        }
    }
    
    /**
     * 
     * @描述：检查元素在集合中是否（可能）存在
     * @作者：严磊
     * @时间：2020年5月31日 下午1:30:24
     * @param key       原始Redis键，会自动加上'bf:'前缀
     * @param element   元素值，字符串类型
     * @return
     */
    public boolean mayExist(String key, String element)
    {
        if ( key == null || element == null )
        {
            throw new RuntimeException("键值均不能为空");
        }
        String actualKey = BF_KEY_PREFIX.concat(key);
        boolean result = false;
        
        /**
         * 使用redis 管道Pipeline，将命令打包推送 参考：https://www.cnblogs.com/yepei/p/5662734.html
         */
        try (Pipeline pipeline = jedis.pipelined())
        {
            for (long index : getBitIndices(element))
            {
                pipeline.getbit(actualKey, index);
            }
            result = !pipeline.syncAndReturnAll().contains(false);
        }
        catch (Exception ex)
        {
            LOGGER.error("资源关闭时发生Exception", ex);
            throw new BloomFilterException(ex);
        }
        return result;
    }
}
